package com.mew.chess.core.common.handler;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.RemovalListener;
import com.github.benmanes.caffeine.cache.Scheduler;
import com.mew.common.rest.bo.GameSession;
import com.zebra.common.concurrent.ZebraThreadFactory;
import io.netty.channel.Channel;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;


public class ChessSessionManager {

    private final Cache<Long, GameSession> sessionCache;
    public static AtomicInteger incrementNumber = new AtomicInteger(0);
    public static AtomicInteger decrementNumber = new AtomicInteger(0);

    public ChessSessionManager(RemovalListener<Long, GameSession> removalListener){
        sessionCache = Caffeine.newBuilder()
                .removalListener(removalListener)
                .expireAfterAccess(120,  TimeUnit.SECONDS)
                .scheduler(Scheduler.forScheduledExecutorService(Executors.newScheduledThreadPool(1, new ZebraThreadFactory("cache-pool"))))
                .build();
    }

    public void put(GameSession session){
        sessionCache.put(session.getUserId(), session);
        incrementNumber.getAndIncrement();
    }

    public void remove(long userId){
        GameSession session = get(userId);
        if (null == session){
            return;
        }
        session.setWeakClose(true);
        sessionCache.invalidate(userId);
        decrementNumber.getAndDecrement();
    }

    public void remove(GameSession chessSession){
        chessSession.setWeakClose(true);
        sessionCache.invalidate(chessSession.getUserId());
        decrementNumber.getAndDecrement();
    }

    public GameSession get(long userId){
        return sessionCache.getIfPresent(userId);
    }


    public List<GameSession> findByChannel(Channel channel){
        return sessionCache.asMap().values().stream().filter(sl -> channel.equals(sl.getChannel())).collect(Collectors.toList());
    }


    public static long size(){
        int onNumber = incrementNumber.get() + decrementNumber.get();
        return onNumber < 0 ? 0 : onNumber;
    }


}
